package com.zjw.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.zjw.shortlink.project.common.constant.RedisKeyConstant;
import com.zjw.shortlink.project.common.constant.ShortLinkConstant;
import com.zjw.shortlink.project.common.convention.exception.ServiceException;
import com.zjw.shortlink.project.config.GotoDomainWhiteListConfiguration;
import com.zjw.shortlink.project.dao.entity.*;
import com.zjw.shortlink.project.dao.mapper.*;
import com.zjw.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.zjw.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.zjw.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import com.zjw.shortlink.project.mq.producer.ShortLinkStatsSaveProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@Slf4j
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "short-link_project-service_topic",
        consumerGroup = "short-link_project-service_stats-save_cg"
)
public class ShortLinkStatsSaveConsumer implements RocketMQListener<Map<String, String>> {
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final StringRedisTemplate stringRedisTemplate;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final ShortLinkMapper shortLinkMapper;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;
    @Override
    public void onMessage(Map<String, String> message) {
        String keys = message.get("keys");
        if(messageQueueIdempotentHandler.isMessageProcessed(keys)) {
            if(messageQueueIdempotentHandler.isAccomplish(keys)) {
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        try {
            String fullShortUrl = message.get("fullShortUrl");
            ShortLinkStatsRecordDTO record = JSON.parseObject(message.get("record"), ShortLinkStatsRecordDTO.class);
            actualSaveStats(fullShortUrl, record);
        } catch(Exception ex) {
            // 处理消息异常时，删除消息已被处理的标识
            messageQueueIdempotentHandler.delMessageProcessed(keys);
            log.error("记录短链接监控消费异常", ex);
        }
        messageQueueIdempotentHandler.setAccomplish(keys);
    }

    public void actualSaveStats(String fullShortUrl, ShortLinkStatsRecordDTO recordDTO) {
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(RedisKeyConstant.LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock lock = readWriteLock.readLock();
        lock.lock();
        try {
            Week week = DateUtil.dayOfWeekEnum(new Date());
            int weekValue = week.getIso8601Value();
            int hour = DateUtil.hour(new Date(), true);
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)
                    .uip(recordDTO.getUipFirstFlag() ? 1 : 0)
                    .uv(recordDTO.getUvFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);
            localeParamMap.put("ip", recordDTO.getRemoteAddr());
            String localeStr = HttpUtil.get(ShortLinkConstant.AMAP_REMOTE_URL, localeParamMap);
            JSONObject localeResult = JSON.parseObject(localeStr);
            String infoCode = localeResult.getString("infocode");
            String actualProvince = "未知";
            String actualCity = "未知";
            if(StrUtil.isNotBlank(infoCode) && Objects.equals(infoCode, "10000")) {
                String province = localeResult.getString("province");
                boolean unkonownFlag = StrUtil.equals(province, "[]");
                actualProvince = unkonownFlag ? actualProvince : province;
                actualCity = unkonownFlag ? actualCity : localeResult.getString("city");
                LinkLocaleStatsDO localeStatsDO = LinkLocaleStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .province(actualProvince)
                        .city(actualCity)
                        .adcode(localeResult.getString("adcode"))
                        .country("中国")
                        .date(new Date())
                        .cnt(1)
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleState(localeStatsDO);
            }
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .os(recordDTO.getOs())
                    .cnt(1)
                    .date(new Date())
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .browser(recordDTO.getBrowser())
                    .cnt(1)
                    .date(new Date())
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .device(recordDTO.getDevice())
                    .cnt(1)
                    .date(new Date())
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .network(recordDTO.getNetwork())
                    .cnt(1)
                    .date(new Date())
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .user(recordDTO.getUv())
                    .browser(recordDTO.getBrowser())
                    .os(recordDTO.getOs())
                    .network(recordDTO.getNetwork())
                    .device(recordDTO.getDevice())
                    .ip(recordDTO.getRemoteAddr())
                    .locale(String.join("-", "中国", actualProvince, actualCity))
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);
            LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = new LambdaQueryWrapper<ShortLinkGotoDO>()
                    .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
            ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
            String gid = shortLinkGotoDO.getGid();
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, recordDTO.getUvFirstFlag() ? 1 : 0, recordDTO.getUipFirstFlag() ? 1 : 0);
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUv(recordDTO.getUvFirstFlag() ? 1 : 0)
                    .todayUip(recordDTO.getUipFirstFlag() ? 1 : 0)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } catch (Exception ex) {
            throw new ServiceException("短链接访问统计异常");
        } finally {
            lock.unlock();
        }
    }
}
